package com.distribute.framework.plugin;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import com.alibaba.fastjson.JSON;
import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.distribute.framework.plugin.Notify.Listener;
import com.distribute.framework.core.web.helper.Utils;

public class AliyunMQ extends BaseMQ{
	public static String defaultTopic = "framework";
	public static Map<String,Producer> producers = new HashMap();
	public static Map<String,Consumer> consumers = new HashMap();
	public static Map<String,List<Listener>> listeners = new HashMap();
	public static Producer getMQProducer(String name){
		if(producers.containsKey(name)){
			return producers.get(name);
		}
		Properties properties = new Properties();		
		properties.put(PropertyKeyConst.ProducerId, Utils.config(name+".broker.ProducerId"));// 您在控制台创建的 Producer ID
        properties.put(PropertyKeyConst.AccessKey, Utils.config(name+".broker.AccessKey"));// AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.SecretKey, Utils.config(name+".broker.SecretKey"));// SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.ONSAddr,Utils.config(name+".broker.url"));//此处以公共云生产环境为例
		Producer producer = ONSFactory.createProducer(properties);
		producers.put(name, producer);
		producer.start();
        return producer;
	}
	private void createConsumer(String name,String topic) {
		Properties properties = new Properties();		
		properties.put(PropertyKeyConst.AccessKey, Utils.config(name+".broker.AccessKey"));// AccessKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.SecretKey, Utils.config(name+".broker.SecretKey"));// SecretKey 阿里云身份验证，在阿里云服务器管理控制台创建
        properties.put(PropertyKeyConst.ONSAddr,Utils.config(name+".broker.url"));//此处以公共云生产环境为例		
		properties.put(PropertyKeyConst.ConsumerId, Utils.config(name+".broker.ConsumerId."+topic));// 您在控制台创建的 Consumer ID
		properties.put(PropertyKeyConst.ConsumeThreadNums,Utils.config(name+".broker.thread", "30"));
		
		Consumer consumer = ONSFactory.createConsumer(properties);
		consumers.put(name+"."+topic, consumer);
		consumer.subscribe(topic, "*", new MessageListener() {
			@SuppressWarnings("unchecked")
			public Action consume(Message message, ConsumeContext consumeContext) {
				try{
					List<Listener> list = listeners.get(message.getTopic()+"."+message.getTag());
					if(list==null){
						return Action.ReconsumeLater;
					}
					for(Listener notifyListener:list){
						try {
							
							String strLocal = Utils.md5(""+notifyListener.hashCode()) ;//监听器的序列
							String strSerial = Utils.or(message.getKey(),"");//消息中传过来的序列
							if(strSerial.indexOf(strLocal)>-1){
								Utils.trace("接收到来自相同监听器发出的消息",strSerial, strLocal,message);
								return Action.CommitMessage;
							}
							Utils.setLocal("serial", strSerial+"["+strLocal+"]");
							notifyListener.execute((new String(message.getBody())));
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
					return Action.CommitMessage;
				}catch(Exception ex){
					Utils.trace("###################消息消费时出错#######################");
					ex.printStackTrace();
					return Action.ReconsumeLater;
				}
			}
        });
		consumer.start();
	}
	
	
	protected static Message setSerial(Message message) {
		String strSerial = Utils.or(message.getKey(), "");
		String strLocal = Utils.getLocal("serial");
		if(!Utils.empty(strLocal) && strSerial.indexOf(strLocal)==-1){
			message.setKey(strSerial+strLocal);
		}
		return message;
	}
	
	public void addMessageListener(String name, String[] tags,final Listener notifyListener, Boolean isP2P) {
		this.addMessageListener(name, defaultTopic, tags, notifyListener, isP2P);
	}
	
	//添加消息监听
	public void addMessageListener(String name,String topic, String[] tags,final Listener notifyListener, Boolean isP2P) {
		createConsumer(name,topic);
		synchronized(listeners){
			for(String tag:tags){
				List<Listener> list = listeners.get(topic+"."+tag);
				if(list==null){
					listeners.put(topic+"."+tag, list = new ArrayList());
				}
				list.add(notifyListener);
			}
		}
	}
	
	//发送文本消息
	public void sendTextMessage(String name,String topic, String tag,String message, Boolean isP2P){
		Message msg = new Message(topic,tag,message.getBytes());
		 setSerial(msg);
		 getMQProducer(name).send(msg);
	}
	
	//发送文本消息
	public void sendTextMessage(String name, String tag,String message, Boolean isP2P) {
		 this.sendTextMessage(name, defaultTopic, tag, message, isP2P);
	}
	
	
	public static void main(String...strings){
		AliyunMQ aliyunMQ = AliyunMQ.getInstance();
		aliyunMQ.sendTextMessage("test", "test", "测试", false);
		Utils.trace("发送成功");
	}
	
	static AliyunMQ aliyunMQ = null;
	public static AliyunMQ getInstance() {
		if(aliyunMQ==null)aliyunMQ = new AliyunMQ();
		return aliyunMQ;
	}
}